import redis
import json
import time
import logging
import os
from threading import Thread
from typing import Dict, Any, Optional
from ths_api import get_trade_manager

# 创建日志目录
log_dir = "logs"
if not os.path.exists(log_dir):
    os.makedirs(log_dir)

# 配置日志 - 使用更详细的配置
logger = logging.getLogger('THSTradeClient')
logger.setLevel(logging.INFO)

# 避免重复添加处理器
if not logger.handlers:
    # 文件处理器
    file_handler = logging.FileHandler(
        os.path.join(log_dir, 'ths_trade_client.log'),
        encoding='utf-8'
    )
    file_handler.setLevel(logging.INFO)

    # 控制台处理器
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)

    # 格式化器
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    file_handler.setFormatter(formatter)
    console_handler.setFormatter(formatter)

    # 添加处理器
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)


class THSTradeClient:
    def __init__(self, strategy_no: str = "1001"):
        # Redis配置 - 增加超时设置
        self.redis_config = {
            'host': '123.60.24.64',
            'port': 6379,
            'password': '@Ym19950928',
            'db': 0,
            'decode_responses': True,
            'socket_connect_timeout': 10,
            'socket_timeout': 30,  # 增加socket超时时间
            'retry_on_timeout': True,  # 超时后自动重试
            'health_check_interval': 30  # 健康检查间隔
        }

        # 使用Streams代替Pub/Sub
        self.order_stream = 'jq_orders_stream'
        self.consumer_group = 'ths_trade_group'
        self.consumer_name = f'consumer_{strategy_no}'

        self.is_running = False
        self.strategy_no = strategy_no
        self.redis_client = None
        self.last_message_id = '0-0'  # 最后处理的消息ID

        # 初始化交易管理器
        self.trade_manager = get_trade_manager(strategy_no)

        logger.info(f"THS交易客户端初始化完成 - 策略编号: {strategy_no}")

    def init_redis(self):
        """初始化Redis连接并创建消费者组"""
        try:
            self.redis_client = redis.Redis(**self.redis_config)
            self.redis_client.ping()
            logger.info("Redis连接成功")

            # 创建消费者组（如果不存在）
            try:
                self.redis_client.xgroup_create(
                    name=self.order_stream,
                    groupname=self.consumer_group,
                    id='0',
                    mkstream=True
                )
                logger.info(f"创建消费者组: {self.consumer_group}")
            except redis.exceptions.ResponseError as e:
                if "BUSYGROUP" in str(e):
                    logger.info(f"消费者组已存在: {self.consumer_group}")
                else:
                    raise

            # 检查并处理积压消息
            self.check_existing_messages()
            return True
        except Exception as e:
            logger.error(f"Redis连接失败: {e}")
            return False

    def check_existing_messages(self):
        """检查并处理积压的消息"""
        try:
            # 检查未处理的消息
            pending_count = self.redis_client.xpending(
                name=self.order_stream,
                groupname=self.consumer_group
            )['pending']

            if pending_count > 0:
                logger.info(f"发现 {pending_count} 条未处理消息，开始处理...")
                self.process_pending_messages()
            else:
                logger.info("没有发现未处理消息")
        except Exception as e:
            logger.error(f"检查消息失败: {e}")

    def process_pending_messages(self):
        """处理未确认的消息"""
        try:
            # 获取未处理的消息
            pending_messages = self.redis_client.xpending_range(
                name=self.order_stream,
                groupname=self.consumer_group,
                min='-',
                max='+',
                count=100,
                consumername=self.consumer_name
            )

            for msg in pending_messages:
                message_id = msg['message_id']
                delivery_count = msg['delivery_count']

                # 获取消息内容
                messages = self.redis_client.xrange(
                    self.order_stream,
                    min=message_id,
                    max=message_id
                )

                if messages:
                    _, message_data = messages[0]
                    order_data = json.loads(message_data['order'])

                    logger.info(f"处理未确认消息 [{message_id}]: {order_data}")

                    # 处理订单
                    result = self.process_order(order_data)
                    if result.get('success'):
                        # 确认消息
                        self.redis_client.xack(
                            self.order_stream,
                            self.consumer_group,
                            message_id
                        )
                        logger.info(f"消息处理成功: {result.get('message', '')}")
                    else:
                        logger.error(f"消息处理失败: {result.get('error', '')}")
                        # 如果多次处理失败，考虑放弃
                        if delivery_count > 3:
                            logger.warning(f"消息 {message_id} 多次处理失败，放弃处理")
                            self.redis_client.xack(
                                self.order_stream,
                                self.consumer_group,
                                message_id
                            )
                else:
                    logger.warning(f"消息 {message_id} 不存在，直接确认")
                    self.redis_client.xack(
                        self.order_stream,
                        self.consumer_group,
                        message_id
                    )

        except Exception as e:
            logger.error(f"处理未确认消息时发生错误: {e}")

    def convert_order_format(self, order_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        将接收到的订单格式转换为内部标准格式

        Args:
            order_data: 原始订单数据

        Returns:
            转换后的标准订单格式
        """
        try:
            # 检查消息类型
            message_type = order_data.get('type', '')
            if message_type != 'order':
                logger.warning(f"忽略非订单消息类型: {message_type}")
                return {}

            # 转换为标准格式
            converted_order = {
                'order_id': order_data.get('order_id'),
                'security': order_data.get('security'),
                'security_name': order_data.get('security_name', ''),
                'amount': order_data.get('amount', 0),
                'price': order_data.get('price'),
                'timestamp': order_data.get('timestamp'),
                'side': 'buy' if order_data.get('is_buy', False) else 'sell',
                'strategy': '聚宽策略',  # 默认策略名称
                'order_type': 'normal'  # 普通订单
            }

            # 验证必要字段
            if not converted_order['order_id'] or not converted_order['security']:
                logger.error(
                    f"订单缺少必要字段: order_id={converted_order['order_id']}, security={converted_order['security']}")
                return {}

            return converted_order

        except Exception as e:
            logger.error(f"转换订单格式失败: {e}")
            return {}

    def process_order(self, order_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        处理订单

        Args:
            order_data: 订单数据

        Returns:
            Dict: 处理结果
        """
        try:
            # 转换订单格式
            converted_order = self.convert_order_format(order_data)
            if not converted_order:
                return {"success": False, "error": "订单格式转换失败"}

            # 验证订单数据
            if not self.validate_order_data(converted_order):
                return {"success": False, "error": "订单数据验证失败"}

            # 记录接收到的订单
            logger.info(f"开始处理订单: {converted_order.get('order_id', '未知')}")

            # 调用交易管理器处理订单
            result = self.trade_manager.process_order_from_redis(converted_order)

            return result

        except Exception as e:
            error_msg = f"处理订单异常: {e}"
            logger.error(error_msg)
            return {"success": False, "error": error_msg}

    def validate_order_data(self, order_data: Dict[str, Any]) -> bool:
        """验证订单数据"""
        try:
            required_fields = ['order_id', 'security', 'side']

            for field in required_fields:
                if field not in order_data:
                    logger.error(f"订单缺少必需字段: {field}")
                    return False

            # 验证证券代码格式
            security = order_data.get('security', '')
            if not security or '.' not in security:
                logger.error(f"证券代码格式错误: {security}")
                return False

            # 验证交易方向
            side = order_data.get('side', '')
            if side not in ['buy', 'sell']:
                logger.error(f"交易方向错误: {side}")
                return False

            # 验证数量
            amount = order_data.get('amount', 0)
            if amount <= 0:
                logger.error(f"订单数量无效: {amount}")
                return False

            return True

        except Exception as e:
            logger.error(f"验证订单数据时发生错误: {e}")
            return False

    def listen_realtime(self):
        """监听实时消息 - 使用Streams模式"""
        logger.info("启动实时消息监听线程")

        while self.is_running:
            try:
                logger.info(
                    f"开始监听Stream: {self.order_stream}，消费者组: {self.consumer_group}，消费者: {self.consumer_name}")

                # 使用XREADGROUP读取消息
                messages = self.redis_client.xreadgroup(
                    groupname=self.consumer_group,
                    consumername=self.consumer_name,
                    streams={self.order_stream: '>'},
                    count=1,
                    block=5000  # 5秒超时
                )

                if messages:
                    # 解析消息
                    stream_name, message_list = messages[0]
                    for message_id, message_data in message_list:
                        try:
                            self.last_message_id = message_id
                            order_json = message_data['order']
                            order_data = json.loads(order_json)
                            logger.info(f"收到实时订单: {order_data.get('order_id', '未知')}")

                            # 在新线程中处理订单
                            thread = Thread(target=self.process_order_thread, args=(order_data, message_id))
                            thread.daemon = True
                            thread.start()

                        except json.JSONDecodeError as e:
                            logger.error(f"JSON解析错误: {e}")
                            # 确认无效消息
                            self.redis_client.xack(
                                self.order_stream,
                                self.consumer_group,
                                message_id
                            )
                        except Exception as e:
                            logger.error(f"处理实时消息时发生错误: {e}")
                else:
                    # 没有消息时进行心跳检查
                    self.check_redis_connection()

            except redis.exceptions.TimeoutError:
                logger.warning("Redis读取超时，重新连接...")
                time.sleep(5)  # 等待5秒后重试
                continue
            except redis.exceptions.ConnectionError as e:
                logger.error(f"Redis连接错误: {e}，尝试重新连接...")
                time.sleep(10)  # 等待10秒后重试
                # 尝试重新初始化Redis连接
                if not self.init_redis():
                    logger.error("Redis重新连接失败")
                    break
                continue
            except Exception as e:
                logger.error(f"监听实时消息失败: {e}，尝试重新连接...")
                time.sleep(10)
                # 尝试重新初始化Redis连接
                if not self.init_redis():
                    logger.error("Redis重新连接失败")
                    break
                continue

        logger.info("实时监听线程已停止")

    def process_order_thread(self, order_data: Dict[str, Any], message_id: str):
        """在新线程中处理订单并确认消息"""
        try:
            result = self.process_order(order_data)
            if result.get('success'):
                logger.info(f"订单处理成功: {result.get('message', '')}")
                # 确认消息
                self.redis_client.xack(
                    self.order_stream,
                    self.consumer_group,
                    message_id
                )
            else:
                logger.error(f"订单处理失败: {result.get('error', '')}")
                # 处理失败时不确认消息，让消息重新入队

            # 记录处理结果
            self.log_order_result(order_data, result)

        except Exception as e:
            logger.error(f"处理订单线程发生错误: {e}")

    def log_order_result(self, order_data: Dict[str, Any], result: Dict[str, Any]):
        """记录订单处理结果"""
        try:
            # 转换订单格式用于记录
            converted_order = self.convert_order_format(order_data)

            log_entry = {
                'timestamp': time.strftime("%Y-%m-%d %H:%M:%S"),
                'order_id': converted_order.get('order_id') if converted_order else order_data.get('order_id'),
                'security': converted_order.get('security') if converted_order else order_data.get('security'),
                'side': converted_order.get('side') if converted_order else (
                    'buy' if order_data.get('is_buy', False) else 'sell'),
                'strategy': converted_order.get('strategy') if converted_order else '聚宽策略',
                'amount': converted_order.get('amount') if converted_order else order_data.get('amount'),
                'price': converted_order.get('price') if converted_order else order_data.get('price'),
                'success': result.get('success', False),
                'message': result.get('message', '') or result.get('error', ''),
                'ths_order_id': result.get('ths_order_id', '')
            }

            # 记录到文件
            log_file = os.path.join("logs", "order_results.log")
            with open(log_file, "a", encoding="utf-8") as f:
                f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")

        except Exception as e:
            logger.error(f"记录订单结果失败: {e}")

    def start_monitoring(self):
        """启动监控线程"""

        def monitor_loop():
            logger.info("启动监控线程")
            while self.is_running:
                try:
                    # 每隔30秒检查一次交易状态
                    status = self.trade_manager.get_trading_status()
                    if status.get('success'):
                        logger.info(
                            f"交易状态检查: API连接={status.get('api_connected', False)}, 最近订单数={status.get('recent_orders', 0)}")
                    else:
                        logger.warning(f"交易状态检查失败: {status.get('error', '')}")

                    # 检查Redis连接状态
                    self.check_redis_connection()

                    time.sleep(30)
                except Exception as e:
                    logger.error(f"监控循环发生错误: {e}")
                    time.sleep(30)

        monitor_thread = Thread(target=monitor_loop, daemon=True)
        monitor_thread.start()

    def check_redis_connection(self):
        """检查Redis连接状态"""
        try:
            if self.redis_client:
                self.redis_client.ping()
            else:
                logger.warning("Redis连接不存在，尝试重新连接")
                self.init_redis()
        except:
            logger.error("Redis连接断开，尝试重连...")
            self.init_redis()

    def start(self):
        """启动客户端"""
        logger.info("开始启动THS交易客户端...")

        if not self.init_redis():
            logger.error("Redis初始化失败，客户端启动失败")
            return False

        self.is_running = True

        # 启动实时监听
        realtime_thread = Thread(target=self.listen_realtime, daemon=True)
        realtime_thread.start()

        # 启动状态监控
        self.start_monitoring()

        logger.info(f"THS交易客户端启动成功 - 策略编号: {self.strategy_no}")
        return True

    def stop(self):
        """停止客户端"""
        logger.info("正在停止THS交易客户端...")
        self.is_running = False
        if self.redis_client:
            try:
                self.redis_client.close()
            except:
                pass
        logger.info("客户端已停止")

    def get_status(self) -> Dict[str, Any]:
        """获取客户端状态"""
        try:
            base_status = {
                'is_running': self.is_running,
                'strategy_no': self.strategy_no,
                'redis_connected': self.redis_client is not None,
                'last_message_id': self.last_message_id,
                'timestamp': time.strftime("%Y-%m-%d %H:%M:%S")
            }

            # 获取交易管理器状态
            trade_status = self.trade_manager.get_trading_status()
            base_status.update(trade_status)

            return base_status

        except Exception as e:
            logger.error(f"获取状态失败: {e}")
            return {
                'is_running': self.is_running,
                'error': str(e),
                'timestamp': time.strftime("%Y-%m-%d %H:%M:%S")
            }


def main():
    # 可以通过命令行参数指定策略编号
    import sys
    strategy_no = sys.argv[1] if len(sys.argv) > 1 else "1001"

    logger.info(f"启动THS交易客户端，策略编号: {strategy_no}")

    client = THSTradeClient(strategy_no=strategy_no)

    if client.start():
        try:
            # 保持主线程运行
            logger.info("客户端运行中，按 Ctrl+C 退出...")
            while client.is_running:
                time.sleep(1)
        except KeyboardInterrupt:
            logger.info("收到中断信号")
        finally:
            client.stop()
    else:
        logger.error("客户端启动失败")


if __name__ == "__main__":
    main()